﻿using System.Text;
using Devonline.Core;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace Devonline.MessageQueue.RabbitMQ;

/// <summary>
/// RabbitMQ 消息队列服务的直接接口
/// </summary>
public interface IRabbitMQService : IMessageQueueService;

/// <summary>
/// RabbitMQ 消息队列服务实现
/// </summary>
/// <param name="logger"></param>
/// <param name="endpoint"></param>
public sealed class RabbitMQService(ILogger<RabbitMQService> logger, IRabbitMQEndpoint endpoint) : IRabbitMQService, IMessageQueueService
{
    private readonly ILogger<RabbitMQService> _logger = logger;
    private readonly IRabbitMQEndpoint _endpoint = endpoint;
    private IConnection? _connection;
    private IChannel? _channel;

    /// <summary>
    /// 启动 RabbitMQ 客户端
    /// </summary>
    /// <returns></returns>
    public async Task StartAsync()
    {
        ArgumentNullException.ThrowIfNullOrWhiteSpace(_endpoint.Host);
        ConnectionFactory factory;
        if (!string.IsNullOrWhiteSpace(_endpoint.UserName) && !string.IsNullOrWhiteSpace(_endpoint.Password) && !string.IsNullOrWhiteSpace(_endpoint.Path))
        {
            factory = new ConnectionFactory
            {
                UserName = _endpoint.UserName,
                Password = _endpoint.Password,
                VirtualHost = _endpoint.Path,
                HostName = _endpoint.Host
            };
        }
        else
        {
            factory = new ConnectionFactory
            {
                Uri = new Uri(_endpoint.Host)
            };
        }

        _connection = await factory.CreateConnectionAsync();
        _channel = await _connection.CreateChannelAsync();
    }
    /// <summary>
    /// 停止全部客户端连接
    /// </summary>
    /// <returns></returns>
    public async Task StopAsync()
    {
        if (_channel is not null && _connection is not null)
        {
            await _channel.CloseAsync();
            await _connection.CloseAsync();
            await _channel.DisposeAsync();
            await _connection.DisposeAsync();
        }
    }
    /// <summary>
    /// 停止订阅或者发布一个 topic 对应的生产者或者消费者
    /// </summary>
    /// <param name="topic"></param>
    /// <returns></returns>
    /// <exception cref="NotImplementedException"></exception>
    public Task StopAsync(string topic)
    {
        throw new NotImplementedException();
    }

    /// <summary>
    /// 发布消息到指定的 topic
    /// </summary>
    /// <typeparam name="TMessage"></typeparam>
    /// <param name="topic">发布主题</param>
    /// <param name="message">消息内容</param>
    /// <returns></returns>
    public async Task PublishAsync<TMessage>(string topic, TMessage message)
    {
        ArgumentNullException.ThrowIfNull(_channel);
        ArgumentNullException.ThrowIfNull(message);

        byte[] body;
        var props = new BasicProperties
        {
            DeliveryMode = DeliveryModes.Persistent
        };

        if (message is IDataSerialize serialize)
        {
            body = serialize.Serialize();
            props.ContentType = ContentType.Txt;
        }
        else
        {
            body = Encoding.UTF8.GetBytes(message.ToJsonString());
            props.ContentType = ContentType.Json;
        }

        await _channel.BasicPublishAsync(_endpoint.Exchange, topic, true, props, body);
        _logger.LogInformation($"已发布主题: {topic} 消息: {message}");
    }
    /// <summary>
    /// 订阅指定 topic 的消息
    /// </summary>
    /// <typeparam name="TMessage"></typeparam>
    /// <param name="topic"></param>
    /// <param name="subscriptionName"></param>
    /// <param name="onSubscribe"></param>
    /// <param name="subscriptionType"></param>
    /// <returns></returns>
    /// <exception cref="NotImplementedException"></exception>
    public async Task SubscribeAsync<TMessage>(string topic, string subscriptionName, Func<TMessage, Task> onSubscribe, string? subscriptionType = null) where TMessage : new()
    {
        ArgumentNullException.ThrowIfNull(_channel);

        var consumer = new AsyncEventingBasicConsumer(_channel);
        consumer.ReceivedAsync += async (model, ea) =>
        {
            var body = ea.Body.ToArray();
            TMessage message = new TMessage();
            if (message is IDataSerialize serialize)
            {
                serialize.Deserialize(body);
            }
            else
            {
                var json = Encoding.UTF8.GetString(body);
                ArgumentNullException.ThrowIfNullOrWhiteSpace(json);
                message = json.ToJsonObject<TMessage>()!;
            }

            _logger.LogInformation("接收到主题: {topic} 的消息: {message}", topic, message.ToJsonString());
            await onSubscribe(message);
        };

        await _channel.BasicConsumeAsync(queue: topic, autoAck: true, consumer: consumer);
        _logger.LogInformation($"已订阅主题: {topic} 的消息");
    }
    /// <summary>
    /// 订阅文本消息
    /// </summary>
    /// <param name="topic"></param>
    /// <param name="subscriptionName"></param>
    /// <param name="onSubscribe"></param>
    /// <param name="subscriptionType"></param>
    /// <returns></returns>
    public async Task SubscribeAsync(string topic, string subscriptionName, Func<string, Task> onSubscribe, string? subscriptionType = null)
    {
        ArgumentNullException.ThrowIfNull(_channel);

        var consumer = new AsyncEventingBasicConsumer(_channel);
        consumer.ReceivedAsync += async (model, ea) =>
        {
            var body = ea.Body.ToArray();
            var message = Encoding.UTF8.GetString(body);
            _logger.LogInformation("接收到主题: {topic} 的消息: {message}", topic, message);
            await onSubscribe(message);
        };

        await _channel.BasicConsumeAsync(queue: topic, autoAck: true, consumer: consumer);
        _logger.LogInformation($"已订阅主题: {topic} 的消息");
    }
}